package com.education.realtime.app.dwd;


import com.education.realtime.app.BaseSQLApp;
import com.education.realtime.common.Constant;
import com.education.realtime.util.SqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

// 互动域 用户课程评价
// ods_db中 过滤 table='comment_info' and type = 'insert'
public class DwdInteractionReview extends BaseSQLApp {
    public static void main(String[] args) {
        new DwdInteractionReview().init(3016,"DwdInteractionReview");
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        readOdsDb(tEnv,"DwdInteractionReview");
        Table reviewTable = tEnv.sqlQuery("select " +
                " data['id'] id," +
                " data['user_id'] user_id," +
                " data['course_id'] course_id," +
                " data['review_txt'] review_txt," +
                " data['review_stars'] review_stars," +
                " date_format(data['create_time'],'yyyy-MM-dd') date_id," +
                " data['create_time'] create_time," +
                " ts" +
                " from ods_db" +
                " where `table` = 'review_info'" +
                " and `type` = 'insert'");
        tEnv.createTemporaryView("review_table",reviewTable);


        Table courseTable = tEnv.sqlQuery("select " +
                " data['id'] id," +
                " data['course_name'] course_name" +
                " from ods_db" +
                " where `table` = 'course_info'" +
                " and `type` = 'bootstrap-insert' "
        );
        tEnv.createTemporaryView("course_table",courseTable);

        Table resultTable = tEnv.sqlQuery("select " +
                "  review.id," +
                "  review.user_id," +
                "  review.course_id," +
                "  course.course_name," +
                "  review.review_txt," +
                "  review.review_stars," +
                "  review.date_id," +
                "  review.create_time," +
                "  review.ts" +
                " from review_table review" +
                " join course_table course" +
                " on review.course_id = course.id");

        tEnv.executeSql("create table "+ Constant.DWD_INTERACTION_REVIEW+"(" +
                " id string," +
                " user_id string," +
                " course_id string," +
                " course_name string," +
                " review_txt string," +
                " review_stars string," +
                " date_id string," +
                " create_time string," +
                " ts string)" + SqlUtil.getKafkaWriteDDL(Constant.DWD_INTERACTION_REVIEW));
        resultTable.executeInsert(Constant.DWD_INTERACTION_REVIEW);
    }
}
